package com.group1.realtime.app.dwd.db;

import com.group1.realtime.app.BaseSQLApp;
import com.group1.realtime.common.Constant;
import com.group1.realtime.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

public class DWD_TradePaymentSuc extends BaseSQLApp {

    public static void main(String[] args) {
        new DWD_TradePaymentSuc().init(
                2021,
                2,
                "DWD_TradePaymentSuc"
        );
    }

    @Override
    protected void handle(StreamExecutionEnvironment env, StreamTableEnvironment tEnv) {
        tEnv.getConfig().setIdleStateRetention(Duration.ofMinutes(15));
        readOdsDb(tEnv,"DWD_TradePaymentSuc");

        Table paymentInfo = tEnv.sqlQuery("select " +
                " `data`['id'] id, " +
                " `data`['order_id'] order_id, " +
                " `data`['total_amount'] total_amount, " +
                " `data`['payment_type'] payment_type, " +
                " `data`['payment_status'] payment_status, " +
                " ts " +
                "from ods_db " +
                "where `database`='edu' " +
                "and `table`='payment_info' " +
                "and `type`='update' " +
                "and `data`['callback_time'] is not null" );
        tEnv.createTemporaryView("payment_info", paymentInfo);

        Table orderInfo = tEnv.sqlQuery("select " +
                " `data`['id'] id, " +
                " `data`['user_id'] user_id, " +
                " ts " +
                "from ods_db " +
                "where `database`='edu' " +
                "and `table`='order_info' " );
        tEnv.createTemporaryView("order_info",orderInfo);

        Table result = tEnv.sqlQuery("select " +
                " pi.id, " +
                " pi.order_id, " +
                " oi.user_id, " +
                " pi.total_amount, " +
                " pi.payment_type, " +
                " pi.payment_status," +
                " pi.ts " +
                " from payment_info pi join order_info oi " +
                " on pi.order_id = oi.id");

        tEnv.executeSql("create table dwd_trade_payment_suc(" +
                "id string," +
                "order_id string," +
                "user_id string," +
                "total_amount string," +
                "payment_type string," +
                "payment_status string," +
                "ts string " +
                ")" + SQLUtil.getKafkaSinkDDL(Constant.TOPIC_DWD_TRADE_PAYMENT_SUC));

        result.executeInsert("dwd_trade_payment_suc");

    }
}
